package com.hliushi.spark.rdd

import org.apache.commons.lang3.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/12 23:16
 */
// 为了使用测试类, 把这个object改成class
class WordCount {

  val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("spark_context")
  val sc = new SparkContext(conf)

  def main(args: Array[String]): Unit = {
    // 1.创建SparkContext
    val conf = new SparkConf().setAppName("word_count")
    val sc = SparkContext.getOrCreate(conf)

    // 2.加载文件
    //    2.1.准备文件
    //    2.2.读取文件
    /** RDD 特点:
     * 1.RDD是数据集
     * 2.RDD是编程模型
     * 3.RDD相互之间有依赖关系
     * 4.RDD是可以分区的
     */
    //val rdd1 = sc.textFile("dataset/wordcount.txt")
    val rdd1 = sc.textFile("hdfs:///wc/input/input.txt")

    // 3.处理数据
    //    3.1.把整句话拆分多个单词
    val rdd2 = rdd1.flatMap(x => x.split(","))
    //    3.2.把每个单词指定一个词频i
    val rdd3 = rdd2.map(x => x -> 1)
    //    3.3.聚合
    val rdd4 = rdd3.reduceByKey((curr, agg) => curr + agg)


    // 4.得到结果
    val result = rdd4.collect()

    result.foreach(println(_))
  }

  @Test
  def sparkContext(): Unit = {
    // 1.Spark Context 如何编写
    //    1.1. 创建SparkConf()
    //        如何需要本地运行的,填写local[6] 6代表线程数 setMaster("local[2]")
    val conf = new SparkConf().setMaster("local[6]").setAppName("spark_context")
    //    1.2. 创建SparkContext
    val sc = new SparkContext(conf)

    // SparkContext身为大入口API, 应该能够创建RDD, 并且设置参数, 设置Jar包..


    // 2.关闭SparkContext, 释放集群资源
  }

  /**
   * 从本地集合创建
   */
  @Test
  def rddCreateLocal(): Unit = {
    val seq = Seq("hello01", "hello02", "hello03")
    val seqDouble = Seq(1.0f, 3.0f, 2.0f)
    // 将本地Scala集合分发到RDD
    val rdd1 = sc.parallelize(seq, 2)
    val rddDouble = sc.parallelize(seqDouble, 2)
    // 将本地Scala集合分发到RDD
    val rdd2 = sc.makeRDD(seq, 2)
  }

  /**
   * 从文件创建
   * 函数原型
   * def textFile(
   * .    path: String,
   * .    minPartitions: Int = defaultMinPartitions): RDD[String]
   * textFile: 返回格式是RDD[String]. 返回的就是文本内容, RDD中每一个元素对应一行数据
   */
  @Test
  def rddCreateFile(): Unit = {
    sc.textFile("hdfs://node01:8020")

    // 1.textFile传入的是什么?
    //    * 传入的是一个路径, 读取路径
    //    * hdfs://  file://   /.../... [这种方式分为在集群中执行和本地执行, 集群执行读取hdfs,本地执行读取文件系统]
    // 2.是否支持分区?
    //    * 假如传入的path是 hdfs:///....
    //    * 分区是由HDFS中文件的block决定的
    // 3.支持什么平台
    //    * 支持Hadoop的几乎所有数据格式, 支持HDFS的访问
    //    * 第三方的支持,可以访问aws和阿里云oos中的文件
    // 4.访问方式
    //    * 支持访问文件夹         例如: sc.textFile("hdfs:///dataset")
    //    * 支持访问压缩文件       例如: sc.textFile("hdfs:///dataset/words.gz")
    //    * 支持通过通配符访问     例如: sc.textFile("hdfs:///dataset/*.txt")
    // 注: 如果使用 file:////... 形式访问本地文件的话,
    //     要确保所有的Worker中对应路径上有这个文件, 则否可能会报错无法找到文件
  }

  /**
   * 从RDD衍生
   */
  @Test
  def rddCreateFromRDD(): Unit = {
    val rdd1 = sc.parallelize(Seq(1, 2, 3))
    // 通过在rdd上执行算子操作, 生成新的rdd
    // 原地计算 ?
    // str.substr 返回新的字符串, 非原地计算
    // 和字符串中的方式很像, 字符串是可变的吗 ? 不可变 final修饰的
    // RDD可变吗 ? 不可变
    val rdd2 = rdd1.map((x: Int) => x + 1)
  }


  /**
   * 函数原型
   * .      def map[U: ClassTag](f: T => U): RDD[U]
   * 作用:
   * .    把RDD中的数据一对一的转为另一种形式
   * 注意点:
   * .    Map是一对一, 如果函数式String -> Array[String], 则新的RDD中每条数据就是一个数组
   */
  @Test
  def mapTest(): Unit = {
    // 1.创建RDD
    val rdd1 = sc.parallelize(Seq(1, 2, 3))
    // 2.执行map操作
    val rdd2 = rdd1.map(x => x * 10)
    // 3.得到结果
    val result = rdd2.collect()
    result.foreach(println(_))

    println("-" * 50)
    // 这里照应了 匿名函数 (x: String) => x.split(" ") 是String -> Array[String]
    // 则新的RDD中每条数据就是一个数组
    val stringsSeq = Seq("hadoop spark hive oozie", "kafka flume sqoop  zookeeper", "storm  azkaban hbase")
    val rddMap = stringsSeq.map((x: String) => x.split(" "))

    // 4.关闭sc
    sc.stop()
  }


  /**
   * 函数原型
   * .      def flatMap[U: ClassTag](f: T => List[U]): RDD[U]
   * .      def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]
   * 作用
   * .      FlatMap算子和Map算子类似, 但是FlatMap是一对多
   * 注意点
   * .      flatMap其实就两个操作,是map + flatten, 也就是先转换, 后扁平化
   * flatten扁平化: [将列表中的集合类型(map,set,list)的所有元素,都放在一个列表]
   */
  @Test
  def flatMapTest(): Unit = {
    // 1.创建RDD
    val stringsSeq = Seq("hadoop spark hive oozie", "kafka flume sqoop  zookeeper", "storm  azkaban hbase")
    val rdd1 = sc.parallelize(stringsSeq)

    // 2.处理数据
    val rdd2 = rdd1.flatMap(x => x.split(" "))
      .filter(x => StringUtils.isNotBlank(x))
    // 3.得到结果
    val result = rdd2.collect()
    result.foreach(println(_))
    // 4.关闭sc
    sc.stop()

    /**
     * hadoop
     * spark
     * hive
     * oozie
     * kafka
     * flume
     * sqoop
     * zookeeper
     * storm
     * azkaban
     * hbase
     */
  }


  /**
   * 函数原型
   * .    def reduceByKey(func: (V, V) => V): RDD[(K ,V)]
   * 作用
   * .    首先按照key分组, 接下来把整组的value计算出一个聚合值, 这个操作非常类似于MapReduce中reduce
   * 注意点
   * .    reduceByKey只能作用于key-value型数据, key-value型数据在当前语境中特指Tuple2
   * .    reduceByKey是一个需要Shuffled 的操作
   * .    和其他的Shuffled相比, reduceByKey是高效的, 因为类似MapReduce的,
   * .    在Map端有一个Combiner, 这样I/O的数据变会减少
   */
  @Test
  def reduceByKeyTest(): Unit = {
    // 1.创建RDD
    val stringsSeq = Seq("hadoop spark hive oozie", "kafka flume sqoop zookeeper", "storm azkaban hbase")
    val rdd1 = sc.parallelize(stringsSeq)
    // 2.处理数据
    val rdd2 = rdd1.flatMap(x => x.split(" "))
      .map(x => x -> 1)

    // curr 是当前聚合后的变量, agg是当前进行聚合的元素
    val rdd3 = rdd2.reduceByKey((curr, agg) => curr + agg)

    // 3.得到结果
    val result = rdd3.collect()

    result.foreach(println(_))
    // 4.关闭sc
    sc.stop()
  }
}